package com.atguigu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.ReviewBean;
import com.atguigu.util.ClickHouseUtil_hjy;
import com.atguigu.util.DateFormatUtil_hjy;
import com.atguigu.util.KafkaUtil_wm;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Dws_ReviewWindow {
    public static void main(String[] args) throws Exception {
        // TODO: 2023/3/28 测试开启：事实表分流表   BaseDbApp
        // TODO: 2023/3/28 需求：评价情况，统计周期：当日  统计粒度：课程  指标：用户平均评分  评价用户数  好评率
        // TODO: 2023/3/28 获取 flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism( 1 );
        // TODO: 2023/3/28 读取 kafka 主题  dwd_review_info 的数据 封装为流
        String topic = "dwd_review_info";
        String groupId = "Dws_ReviewWindow";
        DataStreamSource<String> kafkaDS = env.addSource( KafkaUtil_wm.getFlinkKafkaConsumer( topic, groupId ) );
        // TODO: 2023/3/28 过滤数据 并转为 json 格式   未删除的数据转换为 json 格式输出
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap( new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject( value );
                if (value!=null){
                    out.collect( jsonObject );
                }
//                if ("0".equals( jsonObject.getString( "deleted" ) )) {
//                    out.collect( jsonObject );
//                }
            }
        } );
        // TODO: 2023/3/28 提取 watermark
        SingleOutputStreamOperator<JSONObject> jsonWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness( Duration.ofSeconds( 3 ) )
                        .withTimestampAssigner( new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong( "create_time" );
                            }
                        } )
        );
        // TODO: 2023/3/28 按照 course_id 分组
        KeyedStream<JSONObject, String> keyedStream = jsonWithWMDS.keyBy( line -> line.getString( "course_id" ) );
        // TODO: 2023/3/28 主逻辑 转为 JavaBean
        SingleOutputStreamOperator<ReviewBean> javaBeanDS = keyedStream.flatMap( new RichFlatMapFunction<JSONObject, ReviewBean>() {
//            private ValueState<JSONObject> lastValueState;

            //            @Override
//            public void open(Configuration parameters) throws Exception {
//                StateTtlConfig ttlConfig = new StateTtlConfig.Builder( Time.days( 1 ) )
//                        .setUpdateType( StateTtlConfig.UpdateType.OnReadAndWrite )
//                        .build();
//                ValueStateDescriptor<JSONObject> valueState = new ValueStateDescriptor<>( "lastValue-state", JSONObject.class );
//                valueState.enableTimeToLive( ttlConfig );
//                lastValueState = getRuntimeContext().getState( valueState );
//
//            }
// TODO: 2023/3/28 用户平均评分=总星级分/总评价用户数        好评率=好评用户数/总评价用户数
            @Override
            public void flatMap(JSONObject value, Collector<ReviewBean> out) throws Exception {
//                JSONObject exists = lastValueState.value();

                String courseId = value.getString( "course_id" );
                String userId = value.getString( "user_id" );
                String reviewStars = value.getString( "review_stars" );
                long reviewUserCount = 0L;
                long goodReputationCount = 0L;
                if (reviewStars != null) {
                    reviewUserCount = 1L;
                }
                if ("5".equals( reviewStars )) {
                    goodReputationCount = 1;
                }
                out.collect( ReviewBean.builder()
                        .ReviewUserCount( 1L )
                        .GoodReputation( 1L )
                        .Review_Stars( 1 )
                        .Course_ID( courseId )
                        .User_ID( userId )
                        .build() );
            }
        } );

        // TODO: 2023/3/28 开窗聚合
        SingleOutputStreamOperator<ReviewBean> resultDS = javaBeanDS.windowAll( TumblingEventTimeWindows.of( Time.seconds( 10 ) ) )
                .reduce( new ReduceFunction<ReviewBean>() {
                             @Override
                             public ReviewBean reduce(ReviewBean value1, ReviewBean value2) throws Exception {
                                 int SumScore = value1.getReview_Stars() + value2.getReview_Stars();
                                 long sumUserCount = value1.getReviewUserCount() + value2.getReviewUserCount();
                                 long goodReputCount = value1.getGoodReputation() + value2.getGoodReputation();

                                 value1.setReviewUserCount( value1.getReviewUserCount() + value2.getReviewUserCount() );
                                 value1.setReview_Stars( value1.getReview_Stars() + value2.getReview_Stars() );
                                 value1.setAvgScore( (SumScore * 1D / sumUserCount) );
                                 value1.setFavorableRate( goodReputCount * 1D / sumUserCount );
                                 return value1;
                             }
                         },
                        new AllWindowFunction<ReviewBean, ReviewBean, TimeWindow>() {
                            @Override
                            public void apply(TimeWindow window, Iterable<ReviewBean> values, Collector<ReviewBean> out) throws Exception {
                                ReviewBean next = values.iterator().next();
                                next.setTs( System.currentTimeMillis() );
                                next.setStt( DateFormatUtil_hjy.toYmdHms( window.getStart() ) );
                                next.setEdt( DateFormatUtil_hjy.toYmdHms( window.getEnd() ) );
                                out.collect( next );
                            }
                        } );
        // TODO: 2023/3/28 写出到 ClickHouse
        resultDS.print( "dws层课程粒度评价窗口汇总表，即将写入 clickhouse >>>>>>>" );
        resultDS.addSink( ClickHouseUtil_hjy.getClickHouseSink( "insert into dws_review_window values(?,?,?,?,?,?,?,?,?,?)" ) );
        // TODO: 2023/3/28 启动任务
        env.execute( "Dws_ReviewWindow" );
    }
}
